package com.yuanshi.repair

import java.text.SimpleDateFormat
import java.util.{Calendar, Date}
import com.yuanshi.beans.{RepayFinalBean, TmpVintageBean}
import org.apache.spark.broadcast.Broadcast
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{DataFrame, Row, SparkSession}

import scala.collection.mutable.ListBuffer

object Repair {
  def repair(spark: SparkSession, tmp_vintageDF: DataFrame, finalDF: DataFrame, para0: String, para1: String, para2: String, para3: String, para5: String): Unit = {
    import spark.implicits._
    //2.0将数据封装到bean，装到List集合，方便循环操作
    val lst1 = ListBuffer[TmpVintageBean]()
    val tmp_vintageRDD1: RDD[TmpVintageBean] = tmp_vintageDF.where("effect_time>=args(0) and effect_time<=args(1)")
      .flatMap { case Row(order_code: String, effect_time: String, order_type: String, bystages_num: Int, mob: Int, inspect_time: String, max_late_days: Int) =>
        val bean: TmpVintageBean = TmpVintageBean(order_code, effect_time, order_type, bystages_num, mob, inspect_time, max_late_days)
        lst1 += bean
      }.rdd
    val lst1BC: Broadcast[ListBuffer[TmpVintageBean]] = spark.sparkContext.broadcast(lst1)

    val finalRDD2: RDD[RepayFinalBean] = finalDF.map { case Row(bs_order_id: Int, user_id: Int, order_code: String, effect_time: String, order_type: String, bystages_num: Int, plan_id: Int, repay_period: Int, plan_repay_time: String,
    real_repay_time: String, capital: Double, fee: Double, service: Double, repay_status: Int, flag: Int, late_days: String, late_perio: String) => {
      RepayFinalBean(bs_order_id, user_id, order_code, effect_time, order_type, bystages_num, plan_id, repay_period, plan_repay_time,
        real_repay_time, capital, fee, service, repay_status, flag, late_days, late_perio)
    }
    }.rdd

    //3.0获取可以修复的订单
    val lst2 = ListBuffer[(String, Int, Int, Int)]()
    //定义时间格式
    val sf: SimpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:SS")
    val cal = Calendar.getInstance()
    //val lst2BC: Broadcast[ListBuffer[(String, Int, Int, Int)]] = spark.sparkContext.broadcast(lst2)
    val res1: DataFrame = finalRDD2.flatMap(bean2 => {
      val lst1BV: ListBuffer[TmpVintageBean] = lst1BC.value
      //3.1给生效时间加上mob月后得到新日期
      val d1: Date = sf.parse(bean2.real_repay_time)
      val d2: Date = sf.parse(bean2.effect_time)
      cal.setTime(d2)
      cal.add(Calendar.DAY_OF_MONTH, 1)
      val d3: Date = cal.getTime()
      //3.2根据条件挑出可以修复的订单
      lst1BV.flatMap(bean1 => {
        if (bean2.bystages_num == para3 && bean2.order_type == para2 && bean2.repay_period == para5 && bean1.mob == 1 && bean2.effect_time >= para0
          && bean2.flag == null && bean2.effect_time < para1 && d1.before(d3) || bean2.real_repay_time == null
        ) {
          val tp4: (String, Int, Int, Int) = (bean2.order_code, bean2.bs_order_id, bean2.user_id, bean2.repay_period)
          lst2 += (tp4)
        } else {
          List.empty[(String, Int, Int, Int)]
        }
      })
    }).toDF("order_code", "bs_order_id", "user_id", "repay_period")

    res1.createTempView("v_res1")

    val res2DF: DataFrame = spark.sql("select * from v_res1 order by rand() limit limitnum")

    //4.0开始修复逻辑，并写入hive
    val lst3 = ListBuffer[(String, Int, Int, Int)]()
    //    spark.sparkContext.broadcast(lst3)
    res2DF.map { case Row(order_code: String, bs_order_id: Int, user_id: Int, repay_period: Int) =>
      val tp4: (String, Int, Int, Int) = (order_code, bs_order_id, user_id, repay_period)
      lst3 += tp4
    }
    //val lst3BC: Broadcast[ListBuffer[(String, Int, Int, Int)]] = spark.sparkContext.broadcast(lst3)
    val res: DataFrame = finalRDD2.flatMap(bean2 => {
      lst3.map(t => {
        if (bean2.order_code == t._1 && bean2.bs_order_id == t._2 && bean2.user_id == t._3 && bean2.repay_period == t._4) {
          RepayFinalBean(bean2.bs_order_id, bean2.user_id, bean2.order_code, bean2.effect_time, bean2.order_type, bean2.bystages_num, bean2.plan_id, bean2.repay_period, bean2.plan_repay_time,
            bean2.plan_repay_time, bean2.capital, bean2.fee, bean2.service, 1, 1, bean2.late_days, bean2.late_period)
        } else {
          null
        }
      })
    }).toDF("bs_order_id", "user_id", "order_code", "effect_time", "order_type", "bystages_num", "plan_id", "repay_period", "plan_repay_time", "real_repay_time", "capital", "fee", "serviceervice", "repay_status", "flag", "late_days", "late_perio")

    //将结果写入hdfs
    res.write.save("hdfs://cm-master:9000/opt/hive/warehouse/jindiao_v2/report_repay_final_repair")
  }
}